package fun.tan90.easy.binlog.worker.client;

import cn.hutool.core.net.NetUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import fun.tan90.easy.binlog.base.constants.EasyBinlogConstants;
import fun.tan90.easy.binlog.base.model.WorkerConfig;
import fun.tan90.easy.binlog.base.strategy.AbstractStrategyChoose;
import fun.tan90.easy.binlog.worker.property.EasyBinlogWorkerProperties;
import lombok.Getter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.noear.socketd.SocketD;
import org.noear.socketd.cluster.ClusterClientSession;
import org.noear.socketd.transport.client.ClientSession;
import org.noear.socketd.transport.core.Entity;
import org.noear.socketd.transport.core.Reply;
import org.noear.socketd.transport.core.entity.StringEntity;
import org.noear.socketd.transport.core.listener.EventListener;
import org.noear.socketd.utils.IoConsumer;
import org.noear.socketd.utils.RunUtils;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
 * description TODO
 * company 铁人科技
 *
 * @author 陈浩杰
 * @date 2023/12/15 9:48
 */
@Slf4j
@Getter
@Component
public class EbWorker implements ApplicationListener<ApplicationReadyEvent>, DisposableBean {

    private final AtomicBoolean initialized = new AtomicBoolean(false);

    private WorkerConfig config;

    private ClusterClientSession tcpClusterClientSession;

    private String workId;

    private String[] managerRegisterUrls;

    @Resource
    Environment environment;

    @Resource
    EasyBinlogWorkerProperties easyBinlogWorkerProperties;

    @Resource
    AbstractStrategyChoose strategyChoose;

    @SneakyThrows
    @Override
    public void onApplicationEvent(ApplicationReadyEvent applicationReadyEvent) {
        if (initialized.compareAndSet(false, true)) {
            this.workId = workId();
            this.managerRegisterUrls = managerServerUrls();
            this.tcpClusterClientSession = (ClusterClientSession) SocketD.createClusterClient(managerRegisterUrls)
                    .config(c ->
                            c.heartbeatInterval(easyBinlogWorkerProperties.getHeartbeatInterval())
                                    .connectTimeout(easyBinlogWorkerProperties.getConnectTimeout())
                                    .requestTimeout(easyBinlogWorkerProperties.getRequestTimeout()))
                    .listen(new EventListener()
                            .doOnOpen(session -> log.info("EbWorker doOnOpen {}", JSONUtil.toJsonStr(session.handshake().paramMap())))
                            .doOnMessage((session, message) -> {
                                String event = message.event();
                                boolean request = message.isRequest();
                                boolean subscribe = message.isSubscribe();
                                log.info("EbWorker doOnMessage sid [{}], event[{}], data-size [{}] request [{}], subscribe [{}]", message.sid(), event, message.dataSize(), request, message.isSubscribe());
                                if (request || subscribe) {
                                    Entity entity = strategyChoose.chooseAndExecuteResp(event, message);
                                    session.replyEnd(message, entity);
                                } else {
                                    strategyChoose.chooseAndExecute(event, message);
                                }
                            })
                            .doOnClose(session -> {
                                log.warn("EbWorker doOnClose");
                            })
                            .doOnError((session, error) -> {
                                log.error("EbWorker doOnError {}", JSONUtil.toJsonStr(session.handshake().paramMap()), error);
                            }))
                    .openOrThow();
            this.config = getWorkerConfig();
        }
    }

    @Override
    public void destroy() throws Exception {
        tcpClusterClientSession.closeStarting();
        log.info("EbWorker is destroying");
        RunUtils.runAndTry(() -> Thread.sleep(5000));
        tcpClusterClientSession.close();
        log.info("EbWorker destroyed");
    }

    private String workId() {
        Integer port = environment.getProperty("server.port", Integer.class, 8080);
        String host = NetUtil.localIpv4s().stream().filter(n -> NetUtil.isInRange(n, easyBinlogWorkerProperties.getCidr())).findAny().orElse("0.0.0.0");
        return host + ":" + port;
    }

    private String[] managerServerUrls() {
        List<String> managerRegisterUrls = easyBinlogWorkerProperties.getManagerRegisterUrls();
        String token = easyBinlogWorkerProperties.getToken();
        return managerRegisterUrls.stream().map(managerRegisterUrl -> StrUtil.format("{}?@={}&token={}", managerRegisterUrl, workId, token)).toArray(String[]::new);
    }

    /**
     * 从manager获取worker配置
     */
    private WorkerConfig getWorkerConfig() {
        Reply reply = sendAndRequest(EasyBinlogConstants.WORKER_CONFIG_GET, null, 3000);
        return JSONUtil.toBean(reply.dataAsString(), WorkerConfig.class);
    }

    public void send(String event, String message) {
        try {
            tcpClusterClientSession.send(event, message == null ? null : new StringEntity(message));
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public void broadcastSend(String event, String message) {
        List<ClientSession> sessions = tcpClusterClientSession.getSessionAll()
                .stream()
                .filter(ClientSession::isValid)
                .collect(Collectors.toList());
        for (ClientSession session : sessions) {
            try {
                session.send(event, message == null ? null : new StringEntity(message));
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
    }

    public Reply sendAndRequest(String event, String message, long timeout) {
        try {
            return tcpClusterClientSession.sendAndRequest(event, message == null ? null : new StringEntity(message), timeout).await();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public void sendAndSubscribe(String event, String message, long timeout, IoConsumer<Reply> consumer, Consumer<Throwable> onError) {
        try {
            tcpClusterClientSession.sendAndSubscribe(event, message == null ? null : new StringEntity(message), timeout)
                    .thenReply(consumer)
                    .thenError(onError);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

}
